package com.learn.kafka.stream.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * Kafka 生产者配置父类，封装通用配置
 *
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/7 21:13
 * @updateUser: MI
 * @updateTime: 2025/8/7 21:13
 * @updateRemark: 修改内容
 * @version: 1.0
 */
public class ProducerProperties implements KafkaConfiguration {

    // 存储 Kafka 生产者配置
    protected Properties props;

    // 构造方法：初始化默认配置
    public ProducerProperties() {
        props = new Properties();
        initDefaultConfig();
    }

    // 初始化默认配置
    private void initDefaultConfig() {
        // Kafka 集群地址（默认值）
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.17:9092");
        // Key 序列化器
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        // Value 序列化器
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 消息确认机制（1：leader 落盘确认） 0 发出去就确认 | 1 leader 落盘就确认| all(-1) 所有Follower同步完才确认
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        // 发送失败重试次数(异常自动重试次数)
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        // 批量发送大小（默认 16KB）
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 批量发送等待时间（5ms）
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        // 客户端缓冲区大小（默认32M）满了也会触发消息发送
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 获取元数据的最大阻塞时间（3秒）超时后抛出异常
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
    }

    /**
     * 提供获取配置的方法，供子类使用
     *
     * @return Properties
     */
    @Override
    public Properties getProps() {
        return props;
    }

    /**
     * 提供修改配置的方法（灵活覆盖默认值）
     *
     * @param key   配置键（如 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG）
     * @param value 配置值
     */
    @Override
    public void setConfig(String key, String value) {
        props.put(key, value);
    }

    // 新增：创建并返回KafkaProducer实例（消除重复代码）
    public Producer<Integer, String> createProducer() {
        return new KafkaProducer<>(this.getProps());
    }
}
